package com.winit.cloudlink.message.internal.rabbitmq;

import com.alibaba.fastjson.JSON;
import com.winit.cloudlink.common.compress.Compress;
import com.winit.cloudlink.common.exception.CloudlinkException;
import com.winit.cloudlink.common.utils.StringUtils;
import com.winit.cloudlink.config.ApplicationOptions;
import com.winit.cloudlink.config.Metadata;
import com.winit.cloudlink.config.MqServerOptions;
import com.winit.cloudlink.message.*;
import com.winit.cloudlink.message.exception.MessageSendException;
import com.winit.cloudlink.message.exception.QueueNotFoundException;
import com.winit.cloudlink.message.exception.RejectedMessageException;
import com.winit.cloudlink.message.internal.mapping.MappingStrategy;
import com.winit.cloudlink.message.internal.mapping.MappingStrategyFactory;
import com.winit.cloudlink.message.internal.AbstractMessageTemplate;
import com.winit.cloudlink.message.utils.QueueHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.postprocessor.ZipPostProcessor;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class RabbitmqMessageTemplate extends AbstractMessageTemplate {

	private static final Logger logger = LoggerFactory.getLogger(RabbitmqMessageTemplate.class);

	private volatile ConnectionFactory connectionFactory;
	private RabbitAdmin rabbitAdmin;
	private MessageConverter messageConverter;
	private MessageReturnedListener messageReturnedListener;
	private AtomicBoolean started = new AtomicBoolean(false);
	private ReadWriteLock lock = new ReentrantReadWriteLock();

	private boolean messageSizeLimited = Constants.DEFAULT_MESSAGE_SIZE_LIMITED;
	private int messageWarnBytes = Constants.DEFAULT_MESSAGE_WARN_BYTES;
	private int messageMaxBytes = Constants.DEFAULT_MESSAGE_MAX_BYTES;

	private boolean compressEnabled = Constants.DEFAULT_COMPRESSION_ENABLED;
	private Compress.CompressCodec compressCodec = Constants.DEFAULT_COMPRESSION_CODEC;
	private int noncompressMaxByte = Constants.DEFAULT_NO_COMPRESSION_SIZE;

	public RabbitmqMessageTemplate(Metadata metadata) {
		super(metadata);
		ApplicationOptions applicationOptions = metadata.getApplicationOptions();
		messageSizeLimited = applicationOptions.isMessageSizeLimited();
		messageWarnBytes = applicationOptions.getMessageWarnBytes();
		messageMaxBytes = applicationOptions.getMessageMaxBytes();
		compressEnabled = applicationOptions.isCompressEnabled();
		compressCodec = applicationOptions.getCompressCodec();
		noncompressMaxByte = applicationOptions.getNonCompressMaxBytes();
		start();
	}

	@Override
	public void start() {
		if (!started.get()) {
			try {
				lock.writeLock().lock();
				messageConverter = new CloudlinkMessageConverter(metadata);
				connectionFactory = buildConnectionFactory(metadata);
				if (null == rabbitAdmin) {
					rabbitAdmin = new RabbitAdmin(connectionFactory);
				}
				started.set(true);
			} catch (Throwable ex) {
				throw new CloudlinkException("Active rabbitmq template error.", ex);
			} finally {
				lock.writeLock().unlock();
			}
		}
	}

	@Override
	public void active() {

	}

	@Override
	public void deactive() {
	}

	@Override
	public void shutdown() {
		if (started.get()) {
			try {
				lock.writeLock().lock();
				connectionFactory.clearConnectionListeners();
				if (connectionFactory instanceof CachingConnectionFactory) {
					((CachingConnectionFactory) connectionFactory).destroy();
				}
				if (messageSender != null) {
					messageSender.clear();
				}
				if (messageReceiver != null) {
					messageReceiver.clear();
				}
				started.set(false);
			} finally {
				lock.writeLock().unlock();
			}
		}
	}

	public boolean isActived() {
		return started.get();
	}

	protected ConnectionFactory buildConnectionFactory(Metadata metadata) {
		MqServerOptions serverOptions = metadata.getCurrentZone().getMqServerOptions();
		CachingConnectionFactory rabbitConnectionFactory = new CachingConnectionFactory(serverOptions.getHost(),
				serverOptions.getPort());
		rabbitConnectionFactory.setVirtualHost(serverOptions.getVirtualHost());
		rabbitConnectionFactory.setUsername(serverOptions.getUsername());
		rabbitConnectionFactory.setPassword(serverOptions.getPassword());
		rabbitConnectionFactory.setPublisherReturns(true);
		rabbitConnectionFactory.setPublisherConfirms(false);
		return rabbitConnectionFactory;
	}

	@Override
	protected void sendInternal(Message message) {
		try {
			lock.readLock().lock();
			RabbitTemplate template = createRabbitTemplate();
			try {
				ExchangeType exchangeType = message.getHeaders().getExchageType();
				MappingStrategy mappingStrategy = MappingStrategyFactory.buildMappingStrategy(metadata,
						connectionFactory,
						exchangeType);
				String exchangeName = mappingStrategy.getSendExchangeName(null);
				String routingKey = mappingStrategy.getSendRoutingKey(message);
				template.convertAndSend(exchangeName, routingKey, message, new LimitAndCompressMessagePostProcessor());
				logger.info("the message sent done." + message);
				postHandle(message);
			} catch (AmqpConnectException e) {
				logger.error("the connect error.", e);
				if (connectionFactory instanceof SwitchableConnectionFactory) {
					((SwitchableConnectionFactory) connectionFactory).fireConnectError();
					sendInternal(message);
				}
			} catch (AmqpException e) {
				throw new MessageSendException("sent message error.", e);
			}
		} finally {
			lock.readLock().unlock();
		}
	}

	protected String getMessageType(Message message) {
		MessageHeaders headers = (MessageHeaders) message.getHeaders();
		return headers.getMessageType();
	}

	protected void postHandle(Message message) {

	}

	protected void preHandle(Message message) {
		try {
			lock.readLock().lock();
			MessageHeaders headers = (MessageHeaders) message.getHeaders();
			String retryId = headers.getRetryId();
			if (metadata.getApplicationOptions().isValidDestinationExists() && StringUtils.isBlank(retryId)) {
				String messageType = headers.getMessageType();
				String toAppId = headers.getToApp();
				ExchangeType exchangeType = headers.getExchageType();
				String[] zones = headers.getZones();
				MappingStrategy mappingStrategy = MappingStrategyFactory.buildMappingStrategy(metadata,
						connectionFactory,
						exchangeType);
				boolean queueExists = QueueHelper.checkLocalQueueExists(rabbitAdmin, message, mappingStrategy);
				if (!queueExists) {
					String errorMsg = String.format("Could not find the message receiving queue [exchangeType: %s, messageType: %s, toAppId: %s, zone: %s]",
							exchangeType,
							messageType,
							toAppId,
							StringUtils.join(zones, ","));
					logger.error(errorMsg);
					throw new QueueNotFoundException(errorMsg);
				}
			}
		} finally {
			lock.readLock().unlock();
		}
	}

	private RabbitTemplate createRabbitTemplate() {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		rabbitTemplate.setMessageConverter(messageConverter);
		rabbitTemplate.setChannelTransacted(false);
		rabbitTemplate.setMandatory(true);
		rabbitTemplate.setReturnCallback(new ReturnCallback() {

			@Override
			public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
										String exchange, String routingKey) {

				String messageProperties = JSON.toJSONString(message.getMessageProperties());
				Object messageBody = null;
				if (MessageProperties.CONTENT_TYPE_JSON.equals(message.getMessageProperties().getContentType())) {
					try {
						messageBody = new String(message.getBody(), message.getMessageProperties().getContentEncoding());
					} catch (UnsupportedEncodingException e) {
						logger.error("Failed to convert message objects.", e);
					}
				} else {
					// 未知类型体，返回原始对象。
					messageBody = message.getBody();
				}
				logger.error(String.format("returnedMessage: exchange: %s, routingKey: %s, messageProperties: %s, body: %s",
						exchange,
						routingKey,
						messageProperties,
						messageBody));
				if (null != messageReturnedListener) {
					messageReturnedListener.onReturned(message);
				}
			}
		});
		//logger.info("TemplateActived:createRabbitTemplate:" + connectionFactory.getVirtualHost());
		return rabbitTemplate;
	}

	@Override
	public ConnectionFactory getConnectionFactory() {
		return this.connectionFactory;
	}


	@Override
	public void setMessageReturnedListener(MessageReturnedListener listener) {
		this.messageReturnedListener = listener;
	}

	private class LimitAndCompressMessagePostProcessor implements MessagePostProcessor {
		private MessagePostProcessor compressingPostProcessor = new ZipPostProcessor();

		@Override
		public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
			long messageSize = message.getMessageProperties().getContentLength();
			if (messageSize >= messageWarnBytes) {
				logger.warn("The message size [{}] exceeds the warning threshold [{}],the message body [{}]", messageSize, messageWarnBytes, new String(message.getBody()));
			}
			if (messageSizeLimited && messageSize >= messageMaxBytes) {
				throw new RejectedMessageException(String.format("The message is too large to exceed %d bytes and will be rejected to send,message content:%s", messageMaxBytes, new String(message.getBody())));
			}
			if (compressEnabled && messageSize >= noncompressMaxByte) {
				compressingPostProcessor.postProcessMessage(message);
			}
			return message;
		}
	}
}
